package com.zl.tby.kafka;

import com.zl.tby.core.common.R;
import com.zl.tby.model.TbyUavInfo;
import com.zl.tby.model.TbyUavInfoBackups;
import com.zl.tby.model.TbyUavKafka;
import com.zl.tby.service.*;
import com.zl.tby.vo.TbyUavRecordForm;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.noear.solon.Solon;

import java.math.BigDecimal;
import java.time.Duration;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.*;

import static com.zl.tby.core.base.util.Pg.saveUavTrack;

/**
 * kafka监听
 * @author changpeng
 * @version 1.0
 * @since 2024/9/27 13:40
 */
@Slf4j
public class KafkaListener {

    // 声明一个 Map 来存储每个设备的多个参数
    private static Map<String, List<KafkaListener.FlightData>> deviceDataMap = new HashMap<>();

    public static void init(final String kafkaUrl, final String topic) {
        System.out.println("url: " + kafkaUrl + ", topic: " + topic);
        Thread thread = new Thread(new Runnable() {
            @Override
            public void run() {
                System.out.println("开始监听kafka");
                ITbyUavKafkaService service = Solon.context().getBean(ITbyUavKafkaService.class);
                String GROUP = "tby_group1";
                Properties p = new Properties();
                p.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaUrl);
                p.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 1000);
                p.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
                p.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
                p.put(ConsumerConfig.GROUP_ID_CONFIG, GROUP);
                final KafkaConsumer<String, String> consumer = new KafkaConsumer<>(p);
                consumer.subscribe(Collections.singletonList(topic));
                System.out.println("连接成功，开始监听");
                while (true) {
                    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
                    for (ConsumerRecord<String, String> record : records) {
                        String content = record.value();
                        System.out.println("接收到的数据: " + content);
                        try {
                            R<UavNotify> r = service.add(content);
                            if (r.getCode() == 0) {
                                UavNotify notify = r.getData();
                                receivingData(notify);
                            }
                        } catch (Exception ex) {
                            log.error("保存kafka监听数据异常", ex);
                        }
                    }
                }
            }

            // 获取数据
            private void receivingData(UavNotify notify) {
                String deviceId = notify.getDeviceID(); // 获取设备ID
                // 假设 uavNotify 对象已经包含了相关的数据
                List<UavList> uavList = notify.getUav(); // 获取uav列表
                if (uavList != null && !uavList.isEmpty()) {
                    UavList uav = uavList.get(0); // 获取列表中的第一个 UavList
                    List<UavListItem> uavItems = uav.getItem(); // 获取该 UavList 中的 item 列表
                    if (uavItems != null && !uavItems.isEmpty()) {
                        UavListItem uavListItem = uavItems.get(0); // 获取 item 列表中的第一个 UavListItem
                        String posLongitude = uavListItem.getPosLongitude(); // 获取经度
                        String posLatitude = uavListItem.getPosLatitude(); // 获取纬度
                        String attitudeHeading = uavListItem.getAttitudeHeading();
                        if (attitudeHeading == null || attitudeHeading.isEmpty()) {
                            attitudeHeading = "0";
                        }
                        // 将真高从 String 转换为 int 类型
                        int heading = Integer.parseInt(attitudeHeading);

                        // 如果经度或纬度为空，则不处理这条数据，直接退出
                        if (posLongitude == null || posLongitude.isEmpty() || posLatitude == null || posLatitude.isEmpty() || attitudeHeading == null || attitudeHeading.isEmpty()) {
                            return;
                        }

                        // 将经度和纬度转换为 double 类型
                        double longitude = Double.parseDouble(posLongitude);
                        double latitude = Double.parseDouble(posLatitude);


                        // 进入处理map，看之前的map里是否存储，没存储就是起飞，有存储就是正在飞
                        this.calculatedData(deviceId, longitude, latitude, heading);
                    }
                }
            }

            public void calculatedData(String deviceId, double lon,double lat,int currentHeading) {
                ITbyUavInfoBackupsService iTbyUavInfoBackupsService = Solon.context().getBean(ITbyUavInfoBackupsService.class);
                ITbyUavPointService tbyUavPointService = Solon.context().getBean(ITbyUavPointService.class);

                // 定义变量来存储经度和纬度
                double longitude = 0.0;
                double latitude = 0.0;
                double distance = 0.0;
                int time = 0;
                String latLogRecord = "";
                long UAVId = 0;
                String dealLatLogRecord = "";
                //先判断map里是否存在这个设备
                if (deviceDataMap.containsKey(deviceId)) {

                    //如果进入这里面高度还是0，那么证明他没起飞，返回
                    if (currentHeading == 0 && deviceDataMap.get(deviceId).get(0).getDistance() == 0) {
                        return;
                    }

                    //获取一下之前map中的经纬度，计算出来飞行距离，时间增加
                    // 获取该设备对应的FlightData列表
                    List<KafkaListener.FlightData> dataList = deviceDataMap.get(deviceId);

                    // 假设取列表中的第一个FlightData对象来获取经纬度（根据需求可以调整）
                    if (!dataList.isEmpty()) {
                        KafkaListener.FlightData flightData = dataList.get(0); // 这里可以根据需求选择获取的列表项
                        longitude = flightData.getLongitude();
                        latitude = flightData.getLatitude();
                        //map中的总里程
                        distance = flightData.getDistance();
                        //map中的飞行时间
                        time = flightData.getTime();
                        //map中的经纬度字符串
                        latLogRecord = flightData.getLatLogRecord();
                        //map中的飞行机id
                        UAVId = flightData.getUAVId();
                    }

                    //本次飞行的公里数
                    double kilometers = haversine(longitude, latitude, lon, lat);
                    System.out.printf("移动距离: %.6f 公里%n", kilometers);

                    latLogRecord = latLogRecord + "," + lon + " "+ lat;
                    System.out.println("每次拼接后：" + latLogRecord);
                    dealLatLogRecord = latLogRecord;

                    // 找到第一个逗号的位置
                    int commaIndex1 = dealLatLogRecord.indexOf(',');

                    // 如果找到逗号，则从逗号之后的部分开始截取
                    if (commaIndex1 != -1) {
                        // 截取从第一个逗号后面的部分
                        dealLatLogRecord = dealLatLogRecord.substring(commaIndex1 + 1);
                    }

                    // 创建新的FlightData对象包含更新后的信息
                    KafkaListener.FlightData updatedFlightData = new KafkaListener.FlightData(distance + kilometers, time + 1, lon, lat, currentHeading, latLogRecord, UAVId);

                    // 更新列表中的第一个元素或根据需求更新其他元素
                    dataList.set(0, updatedFlightData);

                    // 更新Map中的数据
                    deviceDataMap.put(deviceId, dataList);

                    // 实时更新数据库中backup表
                    TbyUavInfoBackups uavInfoBackups = new TbyUavInfoBackups();
                    uavInfoBackups.setId(updatedFlightData.getUAVId()); //id
                    uavInfoBackups.setDeviceId(deviceId); //设备id
                    uavInfoBackups.setFlightDistance(BigDecimal.valueOf(updatedFlightData.getDistance()));//飞行里程
                    uavInfoBackups.setFlightTime(updatedFlightData.getTime());//飞行时长
                    uavInfoBackups.setLongitudeLatitudeStr(dealLatLogRecord);//经度纬度字符串
                    uavInfoBackups.setCreateTime(LocalDateTime.now());//飞行时间
                    uavInfoBackups.setDerail((short) 1);
                    iTbyUavInfoBackupsService.addOrUpdate(uavInfoBackups);
                    tbyUavPointService.add(deviceId, lon + "," + lat);

                    if (currentHeading == 0) {//结束了

                        ITbyUavInfoService iTbyUavInfoService = Solon.context().getBean(ITbyUavInfoService.class);
                        ITbyUavRecordService tbyUavRecordService = Solon.context().getBean(ITbyUavRecordService.class);

                        //TODO 此处操作数据库，并且清空对应map中的id数据
                        double sumDistance = distance + kilometers;
                        int sumTime = time + 1;

                        //将类型转换为相同类型，方便之后运算
                        BigDecimal sumTimeBigDecimal = new BigDecimal(sumTime);
                        // 将 sumDistance 转换为 BigDecimal
                        BigDecimal sumDistanceBigDecimal = new BigDecimal(sumDistance);

                        System.out.println("开始进行插入数据库操作：" + deviceId);
                        System.out.println("总里程：" + sumDistance);
                        System.out.println("总飞行时间：" + sumTime);
                        System.out.println("总经度纬度：" + latLogRecord);

                        // 找到第一个逗号的位置
                        int commaIndex = latLogRecord.indexOf(',');

                        // 如果找到逗号，则从逗号之后的部分开始截取
                        if (commaIndex != -1) {
                            // 截取从第一个逗号后面的部分
                            latLogRecord = latLogRecord.substring(commaIndex + 1);
                        }

                        //插入自己的备份表
                        TbyUavInfoBackups tbyUavInfoBackups = new TbyUavInfoBackups();
                        tbyUavInfoBackups.setId(UAVId); //设备id
                        tbyUavInfoBackups.setDeviceId(deviceId); //设备id
                        tbyUavInfoBackups.setFlightDistance(BigDecimal.valueOf(sumDistance));//飞行里程
                        tbyUavInfoBackups.setFlightTime(sumTime);//飞行时长
                        tbyUavInfoBackups.setLongitudeLatitudeStr(latLogRecord);//经度纬度字符串
                        tbyUavInfoBackups.setCreateTime(LocalDateTime.now());//飞行时间
                        tbyUavInfoBackups.setDerail((short) 2);
                        iTbyUavInfoBackupsService.addOrUpdate(tbyUavInfoBackups);

                        //插入无人机图层的空间数据
                        saveUavTrack(deviceId,convertToDate(LocalDateTime.now()),latLogRecord);

                        //根据设备id获取设备信息
                        List<TbyUavInfo> infoList = iTbyUavInfoService.selectByDeviceId(deviceId);
                        if (infoList == null || infoList.isEmpty()) {
                            return;
                        } else {
                            Long infoId = infoList.get(0).getId();
                            BigDecimal infoDistance = infoList.get(0).getFlightDistance();//总里程
                            Long infoFlightCount = infoList.get(0).getFlightCount();//总次数
                            BigDecimal infoTime = infoList.get(0).getFlightTime();//总时长

                            //求和
                            BigDecimal finalDistance = infoDistance.add(sumDistanceBigDecimal);//最终总里程
                            Long finalFlightCount = (infoFlightCount != null ? infoFlightCount : 0L) + 1; // 最终总次数
                            BigDecimal finalTime = infoTime.add(sumTimeBigDecimal);//最终总时长

                            //修改无人机信息
                            iTbyUavInfoService.updateByDeviceId(infoId, finalDistance, finalTime, finalFlightCount);

                            // 记录飞行任务
                            TbyUavRecordForm record = new TbyUavRecordForm();
                            record.setUavId(infoId);
                            record.setFlightTime(sumTimeBigDecimal);
                            record.setFlightDistance(sumDistanceBigDecimal);
                            record.setUseDate(new Date(System.currentTimeMillis()));
                            record.setTaskCode("0");
                            tbyUavRecordService.add(record);

                            removeDataByDeviceId(deviceId);
                        }
                    }
                } else {
                    // 设备不存在，创建一个新的FlightData对象，并初始化飞行时间和飞行距离为0

                    // 使用 String.format() 保留六位小数
                    latLogRecord = String.format("%.6f %.6f", lon, lat);
                    System.out.println("初始化：" + latLogRecord);

                    //定义一个无人机id
                    UAVId = generateId();

                    KafkaListener.FlightData newData = new KafkaListener.FlightData(0.0, 0, lon, lat, currentHeading, latLogRecord, UAVId); // 假设FlightData构造函数接收距离和时间

                    // 创建一个列表来存储FlightData
                    List<KafkaListener.FlightData> dataList = new ArrayList<>();
                    dataList.add(newData);

                    // 将设备ID和数据列表存入Map中
                    deviceDataMap.put(deviceId, dataList);

                    // 输出确认信息
                    System.out.println("已为设备 " + deviceId + " 添加新数据: " + newData);
                }
            }

            //计算飞行距离
            private static double haversine(double lon1, double lat1, double lon2, double lat2) {
                final int R = 6371; // 地球半径，单位为公里

                double dLat = Math.toRadians(lat2 - lat1);
                double dLon = Math.toRadians(lon2 - lon1);
                double a = Math.sin(dLat / 2) * Math.sin(dLat / 2) +
                        Math.cos(Math.toRadians(lat1)) * Math.cos(Math.toRadians(lat2)) *
                                Math.sin(dLon / 2) * Math.sin(dLon / 2);
                double c = 2 * Math.asin(Math.sqrt(a));

                return R * c; // 返回公里数
            }

            // 重置操作
            private void removeDataByDeviceId(String deviceId) {
                // 检查设备 ID 是否存在
                if (deviceDataMap.containsKey(deviceId)) {
                    // 删除该设备 ID 对应的数据
                    deviceDataMap.remove(deviceId);
                    System.out.println("删除设备信息: " + deviceId);
                } else {
                    // 如果设备 ID 不存在，不做任何操作
                    System.out.println("未找到设备信息: " + deviceId);
                }
            }

            // 转换方法
            public static Date convertToDate(LocalDateTime localDateTime) {
                // 使用系统默认时区将 LocalDateTime 转换为 Date
                return Date.from(localDateTime.atZone(ZoneId.systemDefault()).toInstant());
            }

            /**
             * 根据时间戳生成id
             * @author AnWeiNan
             * @date 2024/11/11 下午3:51
             */
            public static long generateId() {
                // 获取当前时间戳的毫秒数
                long timestamp = System.currentTimeMillis();

                // 将当前时间戳转换为字符串
                String idStr = String.valueOf(timestamp);

                // 如果 idStr 长度大于 11，则截取最后 11 位
                if (idStr.length() > 11) {
                    idStr = idStr.substring(idStr.length() - 11);
                }

                // 如果 idStr 长度小于 11，在前面补0，确保长度为11
                while (idStr.length() < 11) {
                    idStr = "0" + idStr;
                }

                // 将字符串转换为 long 类型
                long id = Long.parseLong(idStr);

                return id;
            }

        });
        thread.start();
    }

    // 内部类，表示飞行数据
    public static class FlightData {
        private double distance;       // 飞行距离
        private int time;              // 飞行时间（单位：分钟）
        private double longitude;      // 经度
        private double latitude;       // 纬度
        private int altitude;          // 真高
        private String latLogRecord;   // 经纬度坐标字符串（拼接结果）
        private Long UAVId;            // 无人机id

        // 构造函数
        public FlightData(double distance, int time, double longitude, double latitude, int altitude, String latLogRecord, Long UAVId) {
            this.distance = distance;
            this.time = time;
            this.longitude = longitude;
            this.latitude = latitude;
            this.altitude = altitude;
            this.latLogRecord = latLogRecord;
            this.UAVId = UAVId; // 初始化 UAVId
        }

        // 获取飞行距离
        public double getDistance() {
            return distance;
        }

        // 获取飞行时间
        public int getTime() {
            return time;
        }

        // 获取经度
        public double getLongitude() {
            return longitude;
        }

        // 获取纬度
        public double getLatitude() {
            return latitude;
        }

        // 获取真高
        public int getAltitude() {
            return altitude;
        }

        // 获取经纬度坐标字符串
        public String getLatLogRecord() {
            return latLogRecord;
        }

        // 获取无人机 ID
        public Long getUAVId() {
            return UAVId;
        }

        // 设置无人机 ID
        public void setUAVId(Long UAVId) {
            this.UAVId = UAVId;
        }

        @Override
        public String toString() {
            return "FlightData{" +
                    "distance=" + distance +
                    ", time=" + time +
                    ", longitude=" + longitude +
                    ", latitude=" + latitude +
                    ", altitude=" + altitude +
                    ", latLogRecord='" + latLogRecord + '\'' +
                    ", UAVId=" + UAVId +  // 添加 UAVId 到字符串表示
                    '}';
        }
    }

}
